package com.diego.strom.demo.wordcount;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.IRichBolt;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;

/**
 * 2.创建两个bolt来处理Spout发射出的数据
     Spout已经成功读取文件并把每一行作为一个tuple(在Storm数据以tuple的形式传递)发射过来，我们这里需要创建两个bolt分别来负责解析每一行和对单词计数。
     Bolt中最重要的是execute方法，每当一个tuple传过来时它便会被调用。
 * @author Manfred
 *
 */
public class WordNormalizerRichBolt implements IRichBolt {

	
	/**
	 * 
	 */
	private static final long serialVersionUID = 1L;

	 private OutputCollector collector;  

	public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
		this.collector = collector;  
		
	}

	 /**
	 * 这是bolt中最重要的方法，每当接收到一个tuple时，此方法便被调用 
     * 这个方法的作用就是把文本文件中的每一行切分成一个个单词，并把这些单词发射出去（给下一个bolt处理） 
     * 
     **/  
	public void execute(Tuple input) {
		String sentence = input.getString(0);  
        String[] words = sentence.split(" ");  
        for (String word : words) {  
            word = word.trim();  
            if (!word.isEmpty()) {  
                word = word.toLowerCase();  
                // Emit the word  
                List a = new ArrayList();  
                a.add(input);  
                collector.emit(a, new Values(word));  
            }  
        }  
        //确认成功处理一个tuple  
        collector.ack(input);  
		
	}

	
	public void cleanup() {
		
	}

	
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("word"));  
	}

	
	public Map<String, Object> getComponentConfiguration() {
		// TODO Auto-generated method stub
		return null;
	}

}
